
 /**
 * 上海中赢金融信息服务有限公司
 * Copyright (c) 2017, chinazyjr All Rights Reserved.
 */
package com.sys.kafka.demos.level3;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

/**
 * <b>Description：</b> TODO <br/>
 * <b>ClassName：</b> ProducerImpl <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2017年7月11日 上午11:14:54 <br/>
 * <b>@version: </b>  <br/>
 */
public class ProducerImpl {

    private void createProducer() {  
        Properties props = new Properties();
        props.put("bootstrap.servers", "10.0.3.180:9092,10.0.3.181:9092,10.0.3.182:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//        props.put("partitioner.class", "com.xuanyan.hmc.kafka.handler.MyPartition");
       KafkaProducer<Integer, String> kproducer = new KafkaProducer<Integer, String>(props);
       
//       ProducerRecord<Integer, String> record = new ProducerRecord<Integer, String>("test","test23_60");
       
       Future<RecordMetadata> future=kproducer.send(new ProducerRecord<Integer, String>("test","test2"), new Callback() {
        @Override
        public void onCompletion(RecordMetadata metadata, Exception e) {
            System.out.println("----------------------------------------" +metadata);
            if (e != null)
               System.out.println("the producer has a error:" + e.getMessage());
            else {
                System.out.println("The offset of the record we just sent is: " + metadata.offset());
                System.out.println("The partition of the record we just sent is: " + metadata.partition());
            }

        }
    });
       
       try {
        future.get();
    } catch (InterruptedException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    } catch (ExecutionException e1) {
        // TODO Auto-generated catch block
        e1.printStackTrace();
    }
//       System.out.println(future);
       
//       BlockUtils.sleep(3000);
       kproducer.close();
//       int messageNo = 1;
//       while (true)
//       {
//           String messageStr = new String("Message_" + messageNo);
//           System.out.println("Send:" + messageStr);
////           kproducer.send(new ProducerRecord<Integer, String>("test", messageStr));
//           
//           kproducer.send(new ProducerRecord<Integer, String>("test","呵呵"));
//           messageNo++;
//          BlockUtils.sleep(3000);
//       }
       
//       kproducer.send(new ProducerRecord<Integer, String>("test","呵呵"));
//       System.out.println("---sendok---");
//       kproducer.send(record, new Callback() {
//           @Override
//           public void onCompletion(RecordMetadata metadata, Exception e) {
//               System.out.println("----------------------------------------" +metadata);
//               if (e != null)
//                  System.out.println("the producer has a error:" + e.getMessage());
//               else {
//                   System.out.println("The offset of the record we just sent is: " + metadata.offset());
//                   System.out.println("The partition of the record we just sent is: " + metadata.partition());
//               }
//
//           }
//       });
       
       
        
//        return kproducer;
     }  
    
    /**
     * 发送一条消息
     * @param message
     */
    public void send(KafkaProducer<String, String> kproducer,String topic, String message) {
        ProducerRecord<String, String> record;
        record = new ProducerRecord<>(topic, "", message);
        kproducer.send(record, new SendCallback(kproducer,record, 0));
    }

    
    public static void main(String[] args) {
        ProducerImpl producerImpl = new ProducerImpl();
        producerImpl.createProducer();
        
//        producerImpl.send(producer, "test", "尚");
        System.out.println("----------------完-----");
    }
    
    /**
     * producer回调
     */
    static class SendCallback implements Callback {
        ProducerRecord<String, String> record;
        int sendSeq = 0;
        KafkaProducer<String, String> kproducer;
        public SendCallback(KafkaProducer<String, String> kproducer,ProducerRecord<String, String> record, int sendSeq) {
            this.record = record;
            this.sendSeq = sendSeq;
            this.kproducer = kproducer;
        }
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            
            System.out.println("-----------------onCompletion-------------");
            //send success
            if (null == e) {
                String meta = "topic:" + recordMetadata.topic() + ", partition:"
                        + recordMetadata.topic() + ", offset:" + recordMetadata.offset();
                System.out.println("send message success, record:" + record.toString() + ", meta:" + meta);
                return;
            }
            //send failed
            System.out.println("send message failed, seq:" + sendSeq + ", record:" + record.toString() + ", errmsg:" + e.getMessage());
            if (sendSeq < 1) {
                kproducer.send(record, new SendCallback(kproducer,record, ++sendSeq));
            }
        }
    }
}
